package com.ibm.cps.api;

import javax.inject.Named;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;

import com.ibm.cps.kafka.JudgeTenantType;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.jarxs.ServiceResource;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.model.AuthorizationManager;
import com.ibm.cps.model.ServletUtil;
import com.ibm.cps.servlet.Service;
import com.ibm.cps.servlet.TopologyMode;
import com.ibm.cps.servlet.starter.WorkerProcessManager;
import com.ibm.interfaces.IPersistency;
import com.ibm.mongo.MongoPersistency;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.LocalConfig;
import com.ibm.util.exception.CPSException;

@Named
public class ServiceBean implements ServiceResource{

	private static Logger logger = LoggerFactory.getLogger(Service.class);
	private IPersistency persist = null;
	private static WorkerProcessManager messageForwarderManager;
	private static WorkerProcessManager topologyManager;
	private JudgeTenantType judgeTenantType = new JudgeTenantType();
	private AuthorizationManager authManager;
	private IPersistency persistency=null;
	public ServiceBean() throws CPSException {
		persist = MongoPersistency.getInstace();
		persistency = MongoPersistency.getInstace();
		authManager = AuthorizationManager.getInstance();
		messageForwarderManager = WorkerProcessManager.createManagerForMessageForwarder();
		TopologyMode topologyMode = TopologyMode.valueOf(LocalConfig.CPS_MODE);
		logger.info("Current topology mode: " + topologyMode);
		topologyManager = WorkerProcessManager.createWorkerProcessManager(topologyMode);
	}

	@Override
	public Response GetService(String tenantid,String key) {
		if (!authManager.isAllow(tenantid, key)) {
			ObjectNode ret = JSONUtility.newObjectNode("message", "invalid access");
			ret.put("code", 400);
			return Response.ok(ret.toString(),MediaType.APPLICATION_JSON).build();
		}

		//			String result = topologyManager.isAlive(tenantid);
					String result = "";
					String retMsg = ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE,result);
					return Response.ok(retMsg,MediaType.APPLICATION_JSON).build();


	}


	/**
	 * arguments: tenantid serverIp port type
	 */
	@Override
	public Response PutService(String tenantid, String json,String key) {
		String normalTenant = MessageFields.NORMAL_TENANT;
		if (!authManager.isAllow(tenantid, key)) {
			ObjectNode ret = JSONUtility.newObjectNode("message", "invalid access");
			ret.put("code", 400);
			return Response.ok(ret.toString(),MediaType.APPLICATION_JSON).build();
		}

		String retMsg = null;

		try {
			ArrayNode servers = JSONUtility.getArrayFromString(json);
			for (int i = 0; i < servers.size(); i++) {
				ObjectNode server = (ObjectNode) servers.get(i);
				server.put(MessageFields.TENANTID, tenantid);
				if (!server.has(MessageFields.STATUS)) {
					throw new CPSException(ErrorCode.INVALID_INPUT_ERROR,
							"Please provide the value of status.");
				}

				String status = JSONUtility.GetString(MessageFields.STATUS,server);
				persist.putTenantStatus(tenantid, status);
				if (status.equals("start")) {
					logger.info("received start request for tenant: " + tenantid);
					messageForwarderManager.start(tenantid);
					if(judgeTenantType.IsNormalTenants(tenantid))
					{
						if(persistency.getTenantStatus("notFirstTimeToRun"+normalTenant)==null)
						{
//							logger.debug("fail to create normalTenant when register");
//							throw new CPSException(ErrorCode.NULL,
//									"fail to create normalTenant when register.");
							topologyManager.start(normalTenant);
							persistency.putTenantStatus("notFirstTimeToRun"+normalTenant,"exist");
						}
					}else{
						topologyManager.start(tenantid);
					}

					retMsg = ServletUtil.constructReturnMsg(
							ErrorCode.SUCCESS_CODE,
							"Successfully started process topology.");
				} else if (status.equals("stop")) {
					messageForwarderManager.stop(tenantid);
					if(judgeTenantType.IsNormalTenants(tenantid))
					{
						if(persistency.getTenantStatus(normalTenant)==null)
						{
							topologyManager.stop(normalTenant);
						}
					}else{
						topologyManager.stop(tenantid);
					}

					retMsg = ServletUtil.constructReturnMsg(
							ErrorCode.SUCCESS_CODE,
							"Successfully stopped process topology.");
				} else {
					throw new CPSException(ErrorCode.INVALID_REQUEST_CODE,
							"Error server status " + status + " in "
									+ server.toString());
				}
			}

			return Response.ok(retMsg, MediaType.APPLICATION_JSON).build();
		} catch (CPSException e) {
			e.printStackTrace();
			ObjectNode node = JSONUtility.newObjectNode("message",e.getErrorMessage());
			node.put("code", 400);
			return Response.ok(node.toString(), MediaType.APPLICATION_JSON).build();
		}


	}

}
